package com.qf.rabbitmq1.simple;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;

public class ConsumerMain {

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        //服务器所在机器的ip（如果是安装在虚拟机，写虚拟机ip）
        factory.setHost("localhost");
        //amqp协议端口
        factory.setPort(5672);
        //虚拟主机名
        factory.setVirtualHost("cdm");
        //设置虚拟主机对应用户名和密码
        factory.setUsername("cdm");
        factory.setPassword("cdm");

        //获取连接对象（Connection）
        Connection conn = factory.newConnection();
        //获取channel
        Channel channel = conn.createChannel();

        //声明一个队列
        /*
            queue – 队列名称
            durable – 队列是否支持持久化
            exclusive – 是否独占（一次只能连一个生产端），一般设置为非独占
            autoDelete – 是否字段删除，连接断开后自动删除,一般设置为非自动删除
            arguments – 额外参数
         */
        channel.queueDeclare("cdm.simple",true,false,false,null);

        //消费指定队列
        channel.basicConsume("cdm.simple", false, new DefaultConsumer(channel){
            /*
                消息送达
                    String consumerTag: 消费端的唯一标识。一个消费端是唯一不变的
                    Envelope envelope:封装消息路由键、发送端唯一标识、交换机的对象（消息重发时有用）
                    AMQP.BasicProperties properties: 发送端发送过来的一些参数信息
                    byte[] body: 消息体内容
             */
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    System.out.println("消费者标识：" + consumerTag);
                    System.out.println("封装对象:" + envelope);
                    System.out.println("额外参数:" + properties);
                    System.out.println("收到消息：" + new String(body, "utf-8"));

                    System.out.println("------------------------");


                    int i = 1/0;

                    //业务处理成功
                    /*
                        第一个参数：哪条消息
                        第二个参数：是否将之前未被确认的消息一并确认
                     */
                    channel.basicAck(envelope.getDeliveryTag(),false);
                }catch (Exception e){
                    e.printStackTrace();
                    //业务处理失败 -- 将消息重新消费
                    /*
                        第一个参数：哪条消息
                        第二个参数：是否将之前未被确认的消息一并确认
                        第三个参数：是否重新入队. 是--重新消费；否--变成死信
                     */

                    /*
                        这里处理时，一定要控制重新消费的次数，不能无限重试
                            ① 一旦失败，将这条消息要么交给死信处理，要么直接将其存入mysql数据库，后期通过定时任务重新扫描消费
                            ② 控制重试次数，3-5次后，依然失败，存入mysql数据库，后期通过定时任务重新扫描消费
                                   redis: incr(消息id) -- 设置超时时间
                     */
                    channel.basicNack(envelope.getDeliveryTag(),false,false);
                }
            }
        });
    }

}
